package com.sleep.satokendemo.listener;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RStream;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.redisson.api.listener.MessageListener;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author GuoAo
 * @className RstreamTopicListener
 * @date 2025/7/1
 * @description TODO
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class RstreamTopicListener implements MessageListener<String> {

    private final RedissonClient redissonClient;

    @Override
    public void onMessage(CharSequence charSequence, String s) {
        RStream<Object, Object> stream = redissonClient.getStream("log-stream");
        try{
            stream.createGroup("myGroup", StreamMessageId.ALL);
        }catch (Exception e){}
        Map<StreamMessageId, Map<Object, Object>> streamMessageIdMapMap = stream.readGroup("myGroup", "myConsumer", 10, TimeUnit.SECONDS);
        streamMessageIdMapMap.forEach((id, message) -> {
            log.info("messageId:{}, message:{}",id,message);
            stream.ack("myGroup",id);
        });
    }
}
